package com.moli.iov.gateway.server.task;

import com.moli.iov.gateway.server.model.KafkaSendData;
import com.moli.iov.producer.KafkaProducerServer;
import com.moli.iov.util.SpringContextUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author fudh
 * @ClassNmme KafkaSendTask
 * @date 2019/5/16 16:09
 * @Description: TODO
 */
public class KafkaSendTask implements Runnable {

    private KafkaProducerServer kafkaProducerServer;
    private static final Logger logger = LoggerFactory.getLogger(KafkaSendTask.class);

    private KafkaSendData kafkaSendData;
    public KafkaSendTask(KafkaSendData kafkaSendData) {
        this.kafkaProducerServer = SpringContextUtils.getBean("kafkaProducerServer");
        this.kafkaSendData = kafkaSendData;
    }



    @Override
    public void run() {
        String vin = kafkaSendData.getVin();
        String topic = kafkaSendData.getTopic();
        Object payLoad = kafkaSendData.getPayLoad();
        try {
            logger.info("kafka 发送数据开始，vin:{},topic:{}",vin,topic);
            kafkaProducerServer.asynchronousSendMesForTemplate(topic, vin, payLoad);
            logger.info("kafka 发送数据成功，vin:{},topic:{}",vin,topic);
        } catch (Exception e) {
            logger.error("KafkaSendTask kafka数据发送异常",e);
            e.printStackTrace();
        }
    }
}
